// D:/Developments/lmworkspace/snackmq/broker/src/main/java/com/aiwiown/snackmq/broker/handler/PublishMessageHandler.java

package com.aiwiown.snackmq.broker.handler;

// 【新增】导入 MessageStatus 和 Endpoint

import com.aiwiown.snackmq.broker.auth.AuthorizationService;
import com.aiwiown.snackmq.cluster.metadata.MetadataManager;
import com.aiwiown.snackmq.cluster.raft.SnackMQStateMachine;
import com.aiwiown.snackmq.cluster.raft.closeure.SnackMQApplyClosure;
import com.aiwiown.snackmq.cluster.raft.entity.SnackMQRaftLogEntry;
import com.aiwiown.snackmq.cluster.raft.manager.SnackMQRaftNodeManager;
import com.aiwiown.snackmq.common.auth.Action;
import com.aiwiown.snackmq.common.exception.AuthorizationException;
import com.aiwiown.snackmq.common.handler.MessageHandler;
import com.aiwiown.snackmq.common.handler.RequestContext;
import com.aiwiown.snackmq.common.message.Message;
import com.aiwiown.snackmq.common.message.MessageStatus;
import com.aiwiown.snackmq.common.message.MessageType;
import com.aiwiown.snackmq.common.metrics.MessageMetricEmitter;
import com.aiwiown.snackmq.common.protocol.OperationType;
import com.aiwiown.snackmq.management.scaling.AutoScalerService;
import lombok.extern.slf4j.Slf4j;

/**
 * 【最终修复】PublishMessageHandler 用于处理消息发布请求。
 * 此版本增加了正确的 Leader 检查和重定向逻辑，从根本上解决了无限重试循环的问题。
 */
@Slf4j
public class PublishMessageHandler implements MessageHandler {

    private final SnackMQRaftNodeManager dataRaftNodeManager;
    private final AuthorizationService authorizationService;
    private final MessageMetricEmitter metricEmitter;
    private final MetadataManager metadataManager;

    public PublishMessageHandler(SnackMQRaftNodeManager dataRaftNodeManager,
                                 MetadataManager metadataManager,
                                 MessageMetricEmitter metricEmitter,
                                 AutoScalerService autoScalerService,
                                 SnackMQStateMachine dataStateMachine,
                                 AuthorizationService authorizationService) {
        this.dataRaftNodeManager = dataRaftNodeManager;
        this.authorizationService = authorizationService;
        this.metricEmitter = metricEmitter;
        this.metadataManager = metadataManager;
    }


    @Override
    public MessageType getMessageType() {
        return MessageType.PRODUCE_REQUEST;
    }

    @Override
    public void handle(RequestContext ctx, Message clientRequest) {
        try {
            // --- 【最终修复】: 启用 Leader 检查和重定向逻辑 ---
            // 根本原因: 只有数据 Raft 组的 Leader 才能处理写请求。如果当前节点不是 Leader，
            // 它必须将客户端重定向到真正的 Leader，而不是错误地尝试应用日志。
            if (!dataRaftNodeManager.isLeader()) {
                // 【修复】直接调用 RaftNodeManager 中封装好的方法来获取 Leader 的业务地址。
                String leaderBusinessAddress = dataRaftNodeManager.getLeaderAddress();
                if (leaderBusinessAddress != null) {
                    log.debug("Current node is not the data leader. Redirecting client to the true leader at {}", leaderBusinessAddress);
                    // 构建并发送一个包含重定向地址的响应
                    Message redirectResponse = Message.builder()
                            .type(MessageType.RESPONSE)
                            .messageId(clientRequest.getMessageId())
                            .status(MessageStatus.REDIRECT)
                            .property("redirectAddress", leaderBusinessAddress)
                            .build();
                    ctx.writeResponse(redirectResponse);
                    return; // 必须返回，终止后续处理
                }

                // 如果找不到 Leader，则返回一个服务暂时不可用的错误
                log.warn("Current node is not leader, but could not determine the leader's address. Rejecting request.");
                // 【优化】加入 groupId，使错误信息更具诊断性
                String errorMessage = String.format("Service unavailable: No data leader elected yet for group [%s]. Please retry.", dataRaftNodeManager.getGroupId());
                SnackMQApplyClosure.sendErrorResponse(ctx, clientRequest, errorMessage);
                return;
            }

            // --- 如果当前节点是 Leader，则继续执行 ---

            // 1. 授权检查 (逻辑保持不变)
            Message firstMessage = clientRequest.isBatch() && !clientRequest.getMessages().isEmpty()
                    ? clientRequest.getMessages().get(0)
                    : clientRequest;

            String exchangeName = firstMessage.getProperty("exchangeName");
            String resourceTypeForAuth;
            String resourceNameForAuth;

            if (exchangeName != null && !exchangeName.isEmpty()) {
                resourceTypeForAuth = "EXCHANGE";
                resourceNameForAuth = exchangeName;
            } else {
                resourceTypeForAuth = "TOPIC";
                resourceNameForAuth = firstMessage.getTopic();
            }

            if (resourceNameForAuth == null || resourceNameForAuth.isEmpty()) {
                SnackMQApplyClosure.sendErrorResponse(ctx, clientRequest, "Message must have a topic or an exchangeName property.");
                return;
            }
            // 执行授权
            authorizationService.authorize(clientRequest.getProperty("authToken"), Action.PRODUCE, new AuthorizationService.Resource(resourceTypeForAuth, resourceNameForAuth));

            // 2. 将请求直接提交给 Raft 集群
            SnackMQRaftLogEntry logEntry = new SnackMQRaftLogEntry(OperationType.PRODUCE, clientRequest);
            SnackMQApplyClosure closure = new SnackMQApplyClosure(clientRequest, ctx, this.metricEmitter);
            dataRaftNodeManager.apply(logEntry, closure);

        } catch (AuthorizationException e) {
            log.warn("Authorization failed for publish request: {}", e.getMessage());
            SnackMQApplyClosure.sendErrorResponse(ctx, clientRequest, "Permission denied: " + e.getMessage());
        } catch (Exception e) {
            log.error("在处理发布请求时出错: {}", clientRequest, e);
            SnackMQApplyClosure.sendErrorResponse(ctx, clientRequest, "Internal server error during message processing.");
        }
    }
}